/* Copyright (c) 2023 OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL
v2. You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by Wangyunlai on 2023/08/16.
//
#include <vector>
#include <utility>

#include "sql/optimizer/logical_plan_generator.h"

#include "sql/operator/aggr_logical_operator.h"
#include "sql/operator/calc_logical_operator.h"
#include "sql/operator/delete_logical_operator.h"
#include "sql/operator/explain_logical_operator.h"
#include "sql/operator/insert_logical_operator.h"
#include "sql/operator/join_logical_operator.h"
#include "sql/operator/logical_operator.h"
#include "sql/operator/order_logical_operator.h"
#include "sql/operator/predicate_logical_operator.h"
#include "sql/operator/project_logical_operator.h"
#include "sql/operator/table_get_logical_operator.h"
#include "sql/operator/update_logical_operator.h"
#include "sql/stmt/calc_stmt.h"
#include "sql/stmt/delete_stmt.h"
#include "sql/stmt/explain_stmt.h"
#include "sql/stmt/filter_stmt.h"
#include "sql/stmt/insert_stmt.h"
#include "sql/stmt/select_stmt.h"
#include "sql/stmt/stmt.h"
#include "sql/stmt/update_stmt.h"

// using namespace std;

RC LogicalPlanGenerator::create(Stmt *stmt,
                          std::unique_ptr<LogicalOperator> &logical_operator) {
  RC rc = RC::SUCCESS;
  switch (stmt->type()) {
    case StmtType::CALC: {
      CalcStmt *calc_stmt = static_cast<CalcStmt *>(stmt);
      rc = create_plan(calc_stmt, logical_operator);
    } break;

    case StmtType::SELECT: {
      SelectStmt *select_stmt = static_cast<SelectStmt *>(stmt);
      rc = create_plan(select_stmt, logical_operator);
    } break;

    case StmtType::INSERT: {
      InsertStmt *insert_stmt = static_cast<InsertStmt *>(stmt);
      rc = create_plan(insert_stmt, logical_operator);
    } break;

    case StmtType::DELETE: {
      DeleteStmt *delete_stmt = static_cast<DeleteStmt *>(stmt);
      rc = create_plan(delete_stmt, logical_operator);
    } break;

    case StmtType::UPDATE: {
      UpdateStmt *update_stmt = static_cast<UpdateStmt *>(stmt);
      rc = create_plan(update_stmt, logical_operator);
    } break;

    case StmtType::EXPLAIN: {
      ExplainStmt *explain_stmt = static_cast<ExplainStmt *>(stmt);
      rc = create_plan(explain_stmt, logical_operator);
    } break;
    default: {
      rc = RC::UNIMPLENMENT;
    }
  }
  return rc;
}

RC LogicalPlanGenerator::create_plan(
    CalcStmt *calc_stmt, std::unique_ptr<LogicalOperator> &logical_operator) {
  logical_operator.reset(
      new CalcLogicalOperator(std::move(calc_stmt->expressions())));
  return RC::SUCCESS;
}

RC LogicalPlanGenerator::create_plan(
    SelectStmt *select_stmt,
    std::unique_ptr<LogicalOperator> &logical_operator) {
  std::unique_ptr<LogicalOperator> table_oper(nullptr);

  const std::vector<Table *> &tables = select_stmt->tables();
  const std::vector<Field> &all_fields = select_stmt->query_fields();
  for (Table *table : tables) {
    std::vector<Field> fields;
    // 将需要select的属性按照table分类
    for (const Field &field : all_fields) {
      if (0 == strcmp(field.table_name(), table->name())) {
        fields.push_back(field);
      }
    }

    // 相当于包装一下
    std::unique_ptr<LogicalOperator> table_get_oper(
        new TableGetLogicalOperator(table, fields, true /*read only*/));
    // 只有第一个table才会执行到这个if中，其他都是else
    if (table_oper == nullptr) {
      table_oper = std::move(table_get_oper);
    } else {
      JoinLogicalOperator *join_oper = new JoinLogicalOperator;
      join_oper->add_child(std::move(table_oper));
      join_oper->add_child(std::move(table_get_oper));
      table_oper =
          std::unique_ptr<LogicalOperator>(join_oper);
    }
  }

  // 谓词逻辑算子，就是where后面的东西生成算子
  std::unique_ptr<LogicalOperator> predicate_oper;
  RC rc = create_plan(select_stmt->filter_stmt(), predicate_oper);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to create predicate logical plan. rc=%s", strrc(rc));
    return rc;
  }

  std::unique_ptr<LogicalOperator> aggr_oper(
      new AggrLogicalOperator(select_stmt->aggregations()));
  std::unique_ptr<LogicalOperator> order_oper(nullptr);
  if (select_stmt->orders().size() > 0) {
    std::unique_ptr<LogicalOperator> order_by(
        new OrderLogicalOperator(select_stmt->orders()));
    order_oper = std::move(order_by);
  }
  std::unique_ptr<LogicalOperator> project_oper(
      new ProjectLogicalOperator(all_fields));
  bool have_aggregation = false;
  std::vector<AggregateUnit> tmp = select_stmt->aggregations();
  // 判断是否有聚合函数，顺便判断是否合法
  for (int i = 0; i < tmp.size(); i++) {
    if (tmp[i].aggregation != NO_AGGR) {
      if (i != 0 && !have_aggregation)
        return RC::SQL_SYNTAX;
      else
        have_aggregation = true;
    }
  }
  if (have_aggregation) {
    if (predicate_oper) {
      if (table_oper) {
        predicate_oper->add_child(std::move(table_oper));
      }
      if (order_oper) {
        order_oper->add_child(std::move(predicate_oper));
        aggr_oper->add_child(std::move(order_oper));
      } else {
        aggr_oper->add_child(std::move(predicate_oper));
      }
    } else {
      if (table_oper) {
        if (order_oper) {
          order_oper->add_child(std::move(table_oper));
          aggr_oper->add_child(std::move(order_oper));
        } else {
          aggr_oper->add_child(std::move(table_oper));
        }
      }
    }
    logical_operator.swap(aggr_oper);
  } else {
    if (predicate_oper) {
      if (table_oper) {
        predicate_oper->add_child(std::move(table_oper));
      }
      if (order_oper) {
        order_oper->add_child(std::move(predicate_oper));
        project_oper->add_child(std::move(order_oper));
      } else {
        project_oper->add_child(std::move(predicate_oper));
      }
    } else {
      if (table_oper) {
        if (order_oper) {
          order_oper->add_child(std::move(table_oper));
          project_oper->add_child(std::move(order_oper));
        } else {
          project_oper->add_child(std::move(table_oper));
        }
      }
    }
    logical_operator.swap(project_oper);
  }
  return RC::SUCCESS;
}

RC LogicalPlanGenerator::create_plan(
    FilterStmt *filter_stmt,
    std::unique_ptr<LogicalOperator> &logical_operator) {
  std::vector<std::unique_ptr<Expression>> cmp_exprs;
  const std::vector<FilterUnit *> &filter_units = filter_stmt->filter_units();
  // 遍历条件单元，将条件转换成ComparisonExpr之后加入cmp_exprs
  for (const FilterUnit *filter_unit : filter_units) {
    const FilterObj &filter_obj_left = filter_unit->left();
    const FilterObj &filter_obj_right = filter_unit->right();

    std::unique_ptr<Expression> left(
        filter_obj_left.is_attr
            ? static_cast<Expression *>(new FieldExpr(filter_obj_left.field))
            : static_cast<Expression *>(new ValueExpr(filter_obj_left.values)));

    std::unique_ptr<Expression> right(
        filter_obj_right.is_attr
            ? static_cast<Expression *>(new FieldExpr(filter_obj_right.field))
            : static_cast<Expression *>(
                  new ValueExpr(filter_obj_right.values)));

    ComparisonExpr *cmp_expr = new ComparisonExpr(
        filter_unit->comp(), std::move(left), std::move(right));
    cmp_exprs.emplace_back(cmp_expr);
  }

  std::unique_ptr<PredicateLogicalOperator> predicate_oper;
  if (!cmp_exprs.empty()) {
    std::unique_ptr<ConjunctionExpr> conjunction_expr(
        new ConjunctionExpr(ConjunctionExpr::Type::AND, cmp_exprs));
    predicate_oper = std::unique_ptr<PredicateLogicalOperator>(
        new PredicateLogicalOperator(std::move(conjunction_expr)));
  }

  logical_operator = std::move(predicate_oper);
  return RC::SUCCESS;
}

RC LogicalPlanGenerator::create_plan(
    InsertStmt *insert_stmt,
    std::unique_ptr<LogicalOperator> &logical_operator) {
  Table *table = insert_stmt->table();
  //  vector<Value> values(insert_stmt->values(), insert_stmt->values() +
  //  insert_stmt->value_amount());
  std::vector<std::vector<Value>> insertPartList(
      insert_stmt->insertPartList(),
      insert_stmt->insertPartList() + insert_stmt->insertPartNum());

  InsertLogicalOperator *insert_operator =
      new InsertLogicalOperator(table, insertPartList);
  logical_operator.reset(insert_operator);
  return RC::SUCCESS;
}

RC LogicalPlanGenerator::create_plan(
    DeleteStmt *delete_stmt,
    std::unique_ptr<LogicalOperator> &logical_operator) {
  Table *table = delete_stmt->table();
  FilterStmt *filter_stmt = delete_stmt->filter_stmt();
  std::vector<Field> fields;
  for (int i = table->table_meta().sys_field_num();
       i < table->table_meta().field_num(); i++) {
    const FieldMeta *field_meta = table->table_meta().field(i);
    fields.push_back(Field(table, field_meta));
  }
  std::unique_ptr<LogicalOperator> table_get_oper(
      new TableGetLogicalOperator(table, fields, false /*readonly*/));

  std::unique_ptr<LogicalOperator> predicate_oper;
  RC rc = create_plan(filter_stmt, predicate_oper);
  if (rc != RC::SUCCESS) {
    return rc;
  }

  std::unique_ptr<LogicalOperator> delete_oper(new DeleteLogicalOperator(table));

  if (predicate_oper) {
    predicate_oper->add_child(std::move(table_get_oper));
    delete_oper->add_child(std::move(predicate_oper));
  } else {
    delete_oper->add_child(std::move(table_get_oper));
  }

  logical_operator = std::move(delete_oper);
  return rc;
}

RC LogicalPlanGenerator::create_plan(
    UpdateStmt *update_stmt,
    std::unique_ptr<LogicalOperator> &logical_operator) {
  Table *table = update_stmt->table();
  std::vector<UpdateUnit> update_units = update_stmt->update_units();
  FilterStmt *filter_stmt = update_stmt->filter_stmt();
  std::vector<Field> fields;
  for (int i = table->table_meta().sys_field_num();
       i < table->table_meta().field_num(); i++) {
    const FieldMeta *field_meta = table->table_meta().field(i);
    fields.push_back(Field(table, field_meta));
  }
  std::unique_ptr<LogicalOperator> table_get_oper(
      new TableGetLogicalOperator(table, fields, false /*readonly*/));

  std::unique_ptr<LogicalOperator> predicate_oper;
  RC rc = create_plan(filter_stmt, predicate_oper);
  if (rc != RC::SUCCESS) {
    return rc;
  }

  std::unique_ptr<LogicalOperator> update_oper(
      new UpdateLogicalOperator(table, update_units));

  if (predicate_oper) {
    predicate_oper->add_child(std::move(table_get_oper));
    update_oper->add_child(std::move(predicate_oper));
  } else {
    update_oper->add_child(std::move(table_get_oper));
  }

  logical_operator = std::move(update_oper);
  return rc;
}

RC LogicalPlanGenerator::create_plan(
    ExplainStmt *explain_stmt,
    std::unique_ptr<LogicalOperator> &logical_operator) {
  Stmt *child_stmt = explain_stmt->child();
  std::unique_ptr<LogicalOperator> child_oper;
  RC rc = create(child_stmt, child_oper);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to create explain's child operator. rc=%s", strrc(rc));
    return rc;
  }

  logical_operator =
      std::unique_ptr<LogicalOperator>(new ExplainLogicalOperator);
  logical_operator->add_child(std::move(child_oper));
  return rc;
}
